package test;

import io.rsocket.Payload;
import io.rsocket.RSocket;
import io.rsocket.SocketAcceptor;
import io.rsocket.core.RSocketServer;
import io.rsocket.transport.netty.server.TcpServerTransport;
import io.rsocket.util.DefaultPayload;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.publisher.SignalType;
import reactor.core.publisher.Sinks;

import java.util.concurrent.TimeUnit;

public class Server {

    public static void main(String[] args) throws Exception {
        Sinks.Many<Payload> many = Sinks.many().multicast().onBackpressureBuffer();
        Flux<Payload> flux = many.asFlux();

        reactor.netty.tcp.TcpServer tcpServer = reactor.netty.tcp.TcpServer.create()
                .port(10010)
                .doOnConnection((c)-> {
                    System.out.println("On Connection!");
                    c.onTerminate().doFinally(s-> {
                        System.out.println("On onTerminate!");
                    });
                })
                ;

        SocketAcceptor socketAcceptor = (setup, sendingSocket) -> {
            ///throw new RejectedSetupException("reject setup",new RuntimeException("setup "));
            final RSocket responder = new RSocket() {
                @Override
                public Mono<Payload> requestResponse(Payload payload) {
                    return null;
                }

                @Override
                public Flux<Payload> requestStream(Payload payload) {

                    return flux;
                }

                @Override
                public void dispose() {
                    System.out.println("Dispose!");
                }
            };

            responder.onClose().doFinally(s -> {
                many.tryEmitComplete();
            }).subscribe();
            sendingSocket.onClose().doFinally(s-> {
                System.out.println("Sending Socket Close!");
            }).subscribe();
            return Mono.just(responder);
        };

        RSocketServer.create()
                .acceptor(socketAcceptor)
                /*.acceptor(SocketAcceptor.forRequestChannel((publisher) -> {
                    return flux;
                    *//*return Flux.from(publisher).map(Payload::getDataUtf8)
                            .map(s -> "Echo: " + s)
                            .map(DefaultPayload::create);*//*
                }))*/
                .bindNow(TcpServerTransport.create(tcpServer));
        int i = 0;
        while (true) {
            i++;
            many.emitNext(DefaultPayload.create("many.tryEmitNext()" + i), new Sinks.EmitFailureHandler() {
                @Override
                public boolean onEmitFailure(SignalType signalType, Sinks.EmitResult emitResult) {
                    return false;
                }
            });
            // many.tryEmitNext(DefaultPayload.create("many.tryEmitNext()" + i));
            TimeUnit.SECONDS.sleep(1);
        }

       /* synchronized (Server.class) {
            Server.class.wait();
        }*/
    }
}
